package com.codeduck.kafka.event.listener;

import com.codeduck.kafka.handler.IMessageHandler;
import com.codeduck.kafka.event.message.MessageEvent;
import com.codeduck.kafka.event.message.MessageInfo;
import com.codeduck.kafka.pojo.TopicHandler;
import com.codeduck.kafka.util.SpringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * 消息事件监听类：只要监听器对象在 Spring 应用程序上下文中注册，它就会接收事件
 *
 * @author: <a href="mailto:codeduck@163.com">Codeduck</a>
 * @date: 2023/3/7
 */
@Component
public class MessageListener implements ApplicationListener<MessageEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(MessageListener.class);

    @Override
    public void onApplicationEvent(MessageEvent event) {
        MessageInfo messageInfo = event.getMessageInfo();
        if (messageInfo.getMsgHandler().isBatchHandler()) {
            this.batchConsume(messageInfo.getRecords(), messageInfo.getMsgHandler());
        } else {
            this.consume(messageInfo.getRecord(), messageInfo.getMsgHandler());
        }
    }

    private void consume(ConsumerRecord<Object, Object> record, TopicHandler msgHandler) {
        LOGGER.info("消息监听器已接收消息：topic[{}],partition[{}],offset[{}],timestamp[{}]", record.topic(), record.partition(), record.offset(), record.timestamp());
        IMessageHandler handler = SpringUtils.getBean(msgHandler.getHandler());
        handler.consume(record);
    }

    private void batchConsume(ConsumerRecords<Object, Object> records, TopicHandler msgHandler) {
        LOGGER.info("消息监听器已接收消息：topic[{}]，数量[{}]", msgHandler.getTopic(), records.count());
        IMessageHandler handler = SpringUtils.getBean(msgHandler.getHandler());
        handler.batchConsume(records);
    }
}
